package com.bkhech.bigdata.tiktok_anchor.dataclean;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Objects;

/**
 * 实现自定义Map类，在里面实现具体的清洗逻辑
 *
 * @author guowm
 * @date 2022/7/6
 */
public class DataCleanMap extends Mapper<LongWritable, Text, Text, Text> {

    /**
     * 1：从原始数据中过滤出来需要的字段
     * 2：针对核心字段进行异常值判断
     *
     * @param k1 每一行的行首偏移量
     * @param v1 每一行内容
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable k1, Text v1, Mapper<LongWritable, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {
        //  获取到每一行内容
        final String line = v1.toString();
        //  把json字符串数据转成json对象，从json对象中获取需要的字段信息
        final AnchorData anchorData = JSONObject.parseObject(line, AnchorData.class);

        //  过滤掉异常数据
        if (Objects.nonNull(anchorData)
                && StrUtil.isNotBlank(anchorData.getUid())
                // 在做完数据指标统计之后，发现一组 uid == "null" 字符串的异常数据
                // 在此处提前过滤掉 uid == "null" 字符串的异常数据
                && !"null".equals(anchorData.getUid())
                && anchorData.getGold() >= 0
                && anchorData.getWatchNumPv() >= 0
                && anchorData.getFollower() >= 0
                && anchorData.getLength() >= 0) {

            // 组装k2,v2
            Text k2 = new Text();
            k2.set(anchorData.getUid());
            Text v2 = new Text();
            String v2b = anchorData.getGold() + "\t" +
                    anchorData.getWatchNumPv() + "\t" +
                    anchorData.getFollower() + "\t" +
                    anchorData.getLength();
            v2.set(v2b);
            // 写出到 hdfs
            context.write(k2, v2);
        }

    }
}
